package com.example.chatapp.manager;

import com.example.chatapp.redis.RedisMessageSubscriber;
import com.example.chatapp.websocket.WebSocketSessionManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis Pub/Sub 管理器
 * 职责：管理Redis频道订阅，避免Service直接操作RedisMessageListenerContainer
 */
@Component
public class RedisMessageQueueManager {
    
    private final RedisMessageListenerContainer listenerContainer;
    private final WebSocketSessionManager sessionManager;
    private final String channelPrefix;
    
    // 已订阅的频道
    private final Set<String> subscribedChannels = ConcurrentHashMap.newKeySet();
    // 订阅器缓存
    private final Map<String, RedisMessageSubscriber> subscriberCache = new ConcurrentHashMap<>();
    
    public RedisMessageQueueManager(RedisMessageListenerContainer listenerContainer,
                                    WebSocketSessionManager sessionManager,
                                    @Value("${app.redis.redis-channel-prefix}") String channelPrefix) {
        this.listenerContainer = listenerContainer;
        this.sessionManager = sessionManager;
        this.channelPrefix = channelPrefix;
    }
    
    /**
     * 确保订阅指定房间的频道（避免重复订阅）
     */
    public synchronized void ensureSubscribed(String roomId) {
        String channel = channelPrefix + roomId;
        
        // ✅ 双重检查，避免并发问题
        if (subscribedChannels.contains(channel)) {
            System.out.println("[RedisMessageQueueManager] Already subscribed to: " + channel);
            return;
        }
        
        // ✅ 为每个频道创建独立的订阅器实例（共享同一个 sessionManager）
        RedisMessageSubscriber subscriber = subscriberCache.computeIfAbsent(
            channel, k -> {
                System.out.println("[RedisMessageQueueManager] Creating new subscriber for: " + channel);
                return new RedisMessageSubscriber(sessionManager);
            }
        );
        
        // ✅ 添加监听器
        listenerContainer.addMessageListener(subscriber, new ChannelTopic(channel));
        subscribedChannels.add(channel);
        
        System.out.println("[RedisMessageQueueManager] Successfully subscribed to: " + channel + 
                           " (Total subscribers: " + subscribedChannels.size() + ")");
    }
    
    /**
     * 取消订阅
     */
    public void unsubscribe(String roomId) {
        String channel = channelPrefix + roomId;
        subscribedChannels.remove(channel);
        subscriberCache.remove(channel);
        System.out.println("[RedisMessageQueueManager] Unsubscribed from: " + channel);
    }
    
    /**
     * 检查是否已订阅
     */
    public boolean isSubscribed(String roomId) {
        String channel = channelPrefix + roomId;
        return subscribedChannels.contains(channel);
    }
    
    /**
     * 获取已订阅频道列表
     */
    public Set<String> getSubscribedChannels() {
        return Set.copyOf(subscribedChannels);
    }
}
